package com.example.klinemqflink.service;

import com.example.klinemqflink.config.RabbitMQConfig;
import com.example.klinemqflink.job.KlineCalculateService;
import com.example.klinemqflink.model.MarketData;
import com.example.klinemqflink.model.RMDeserializationSchema;
import jakarta.annotation.Resource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;

@Component
public class FlinkRabbitMQReader implements CommandLineRunner {

    @Resource
    KlineCalculateService klineCalculateService;

    @Value("${spring.rabbitmq.host}")
    private String rabbitMQHost;

    @Value("${spring.rabbitmq.port}")
    private int rabbitMQPort;

    @Value("${spring.rabbitmq.username}")
    private String rabbitMQUsername;

    @Value("${spring.rabbitmq.password}")
    private String rabbitMQPassword;

    @Value("${spring.rabbitmq.queue.1}")
    private String rabbitMQQueueName1;

    @Value("${spring.rabbitmq.queue.2}")
    private String rabbitMQQueueName2;

    @Value("${spring.rabbitmq.queue.3}")
    private String rabbitMQQueueName3;

    @Resource
    StreamExecutionEnvironment flinkEnv;

    public void readFromRabbitMQ() throws Exception {
        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = flinkEnv;
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置 RabbitMQ 连接
        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost(rabbitMQHost)
                .setPort(rabbitMQPort)
                .setUserName(rabbitMQUsername)
                .setPassword(rabbitMQPassword)
                .setVirtualHost("/")
                .build();
        // 定义多个队列名称
        List<String> queueNames = RabbitMQConfig.MARKET_DATA_QUEUES;
        DataStream<MarketData> combinedStream = null;

        // 为每个队列创建 RMQSource 并合并数据流
        for (String queueName : queueNames) {
            RMQSource<MarketData> source = new RMQSource<>(
                    connectionConfig,
                    queueName,
                    new RMDeserializationSchema()
            );

            DataStream<MarketData> stream = env.addSource(source);

            if (combinedStream == null) {
                combinedStream = stream;
            } else {
                combinedStream = combinedStream.union(stream);
            }
        }


        // 创建 RabbitMQ 数据源,单个队列
//        DataStream<MarketData> stream = env.addSource(new RMQSource<>(
//                connectionConfig,
//                rabbitMQQueueName,
//                true,
//                new RMDeserializationSchema()
//        ));
        klineCalculateService.processLines(combinedStream);
        // 执行 Flink 作业
        env.execute("Flink RabbitMQ Reader");
    }

    @Override
    public void run(String... args) throws Exception {
//        readFromRabbitMQ();
    }
}
